# -*- coding: utf-8 -*-
from datetime import timedelta
from utils.operators.spark_submit import SparkSubmitOperator

jdbcUrl='{{ var.json.oracle_tab.url }}'
username='{{ var.json.oracle_tab.username }}'
password='{{ var.json.oracle_tab.password }}'
nowdt='{{ execution_date | cst_hour_ds }}'
dtvalue='{{ execution_date | cst_hour }}'
nextdt='{{ execution_date | hour_add(1) | cst_hour_ds }}'
table="yl_oms_oms_order_back"
env='{{ var.value.env_sync }}'

jsonpara="""{
"reader":{
"connect":{
"url":"jdbcUrlpara",
"username":"usernamepara",
"password":"passwordpara",
"driver":"oracle.jdbc.driver.OracleDriver"
},
"dbtype":"oracle",
"tableName":"yl_oms_oms_order",
"indexName":"OMS_OMS_ORDER_IDX2",
"where":"(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and input_time<to_date('nextdt:00:00','yyyy-mm-dd hh24:mi:ss')) ;",
"query":"",
"splitColumn":"waybill_id",
"equalitySectioning":0,
"containsnull":0,
"fetchsize":"1024",
"threadNumber":-1
},
"channel":{
"filterAbnormalCharacter":1
},
"writer":{
"dbtype":"hive",
"tableName":"tablepara",
"database":"jms_ods",
"writeMode": "overwrite",
"partitionColumn":"dt",
"partitionValue":"dtvalue"},
"settting":{
"env":"envpara"}
}""".replace("jdbcUrlpara",jdbcUrl).replace("usernamepara",username).replace("passwordpara",password). \
    replace("nowdt",nowdt).replace("nextdt",nextdt).replace("dtvalue",dtvalue). \
    replace("tablepara",table). \
    replace("envpara",env)

jms_ods__yl_oms_oms_order_hi_input = SparkSubmitOperator(
    task_id='jms_ods__yl_oms_oms_order_hi_input',
    email=['suning@jtexpress.com','yl_bigdata@yl-scm.com'],
    name='jms_ods__yl_oms_oms_order_hi_input_{{ execution_date | hour_add(1) | cst_hour }}',
    pool='oracle_tab',
    pool_slots=1,
    execution_timeout=timedelta(hours=1),
    driver_memory='4G',
    executor_memory='4G',
    executor_cores=2,
    num_executors=2,
    conf={'spark.dynamicAllocation.enabled'                  : 'true',
          'spark.shuffle.service.enabled'                    : 'true',
          'spark.dynamicAllocation.maxExecutors'             : 7,
          'spark.dynamicAllocation.cachedExecutorIdleTimeout': 30,
          'spark.sql.sources.partitionOverwriteMode'         : 'dynamic',
          'spark.executor.memoryOverhead'                    : '2G',
          },
    java_class='com.yunlu.bigdata.jobs.synchrotool.DataSynchDriver',  # spark 主类
    application='hdfs:///scheduler/jms/spark/rgf/spark_sync.jar',  # spark jar 包
    application_args=[jsonpara,],
    sla=timedelta(hours=2),
)


jsonpara_append="""{
"reader":{
"connect":{
"url":"jdbcUrlpara",
"username":"usernamepara",
"password":"passwordpara",
"driver":"oracle.jdbc.driver.OracleDriver"
},
"dbtype":"oracle",
"tableName":"yl_oms_oms_order",
"indexName":"OMS_OMS_ORDER_IDX1",
"where":"
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-38  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-37)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-37  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-36)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-36  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-35)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-35  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-34)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-34  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-33)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-33  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-32)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-32  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-31)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-31  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-30)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-30  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-29)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-29  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-28)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-28  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-27)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-27  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-26)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-26  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-25)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-25  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-24)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-24  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-23)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-23  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-22)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-22  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-21)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-21  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-20)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-20  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-19)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-19  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-18)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-18  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-17)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-17  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-16)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-16  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-15)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-15  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-14)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-14  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-13)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-13  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-12)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-12  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-11)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-11  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-10)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-10  and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-9)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-9   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-8)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-8   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-7)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-7   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-6)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-6   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-5)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-5   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-4)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-4   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-3)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-3   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-2)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-2   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-1)   and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-1   and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss'))     and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-39	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-38)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-40	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-39)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-41	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-40)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-42	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-41)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-43	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-42)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-44	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-43)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-45	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-44)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-46	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-45)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-47	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-46)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-48	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-47)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-49	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-48)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-50	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-49)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-51	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-50)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-52	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-51)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-53	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-52)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-54	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-53)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-55	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-54)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-56	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-55)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-57	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-56)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-58	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-57)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-59	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-58)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-60	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-59)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-61	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-60)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-62	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-61)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-63	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-62)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-64	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-63)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-65	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-64)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-66	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-65)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-67	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-66)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-68	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-67)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-69	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-68)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-70	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-69)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-71	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-70)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-72	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-71)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-73	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-72)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-74	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-73)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-75	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-74)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-76	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-75)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-77	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-76)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-78	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-77)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-79	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-78)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-80	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-79)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-81	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-80)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-82	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-81)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-83	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-82)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-84	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-83)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-85	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-84)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-86	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-85)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-87	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-86)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-88	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-87)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-89	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-88)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
(input_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-90	and input_time<to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss')-89)  and ((update_time>=to_date('nowdt:00:00','yyyy-mm-dd hh24:mi:ss') and update_time<to_date('nextdt:00:00', 'yyyy-mm-dd hh24:mi:ss'))) ;
",
"query":"",
"splitColumn":"waybill_id",
"equalitySectioning":0,
"containsnull":0,
"fetchsize":"1024",
"threadNumber":-1
},
"channel":{
"filterAbnormalCharacter":1
},
"writer":{
"dbtype":"hive",
"tableName":"tablepara",
"database":"jms_ods",
"writeMode": "append",
"partitionColumn":"dt",
"partitionValue":"dtvalue"},
"settting":{
"env":"envpara"}
}""".replace("jdbcUrlpara",jdbcUrl).replace("usernamepara",username).replace("passwordpara",password). \
    replace("nowdt",nowdt).replace("nextdt",nextdt).replace("dtvalue",dtvalue). \
    replace("tablepara",table). \
    replace("envpara",env)

jms_ods__yl_oms_oms_order_hi = SparkSubmitOperator(
    task_id='jms_ods__yl_oms_oms_order_hi',
    email=['suning@jtexpress.com','yl_bigdata@yl-scm.com'],
    name='jms_ods__yl_oms_oms_order_hi_{{ execution_date | hour_add(1) | cst_hour }}',
    pool='oracle_tab',
    pool_slots=1,
    execution_timeout=timedelta(hours=1),
    driver_memory='4G',
    executor_memory='4G',
    executor_cores=2,
    num_executors=2,
    conf={'spark.dynamicAllocation.enabled'                  : 'true',
          'spark.shuffle.service.enabled'                    : 'true',
          'spark.dynamicAllocation.maxExecutors'             : 7,
          'spark.dynamicAllocation.cachedExecutorIdleTimeout': 30,
          'spark.sql.sources.partitionOverwriteMode'         : 'dynamic',
          'spark.executor.memoryOverhead'                    : '2G',
          },
    java_class='com.yunlu.bigdata.jobs.synchrotool.DataSynchDriver',  # spark 主类
    application='hdfs:///scheduler/jms/spark/rgf/spark_sync.jar',  # spark jar 包
    application_args=[jsonpara_append,],
    sla=timedelta(hours=2),
)

jms_ods__yl_oms_oms_order_hi << jms_ods__yl_oms_oms_order_hi_input